package org.ykhl.mq.dispatcher.rule;

import com.alibaba.dubbo.rpc.RpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.ykhl.log.commons.sender.LogSender;
import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.NamedThreadFactory;
import org.ykhl.mq.commons.Utils;
import org.ykhl.mq.dispatcher.rule.model.DispatchRule;
import org.ykhl.mq.dispatcher.MqMessageDispatcher;
import org.ykhl.model.RetInfo;
import org.ykhl.service.ConvertService;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class MqMessageTransformer<T extends BaseMessage> {
	private static final Logger LOG = LoggerFactory.getLogger(MqMessageTransformer.class);

	private Class<T> clazz;
	private final DispatchService dispatchService;

	private final ExecutorService executorService;

	private final ConvertService convertService;
	private final LogSender logSender;
	public MqMessageTransformer(Class<T> clazz, DispatchService dispatchService,ConvertService convertService,LogSender logSender){

		this(clazz, dispatchService,convertService,2 * 8,logSender);
	}

	public MqMessageTransformer(Class<T> clazz, DispatchService dispatchService, ConvertService convertService,int transThreadCnt,LogSender logSender){
		LOG.debug("MqMessageTransformer");
		this.clazz = clazz;

		this.dispatchService = dispatchService;
		this.convertService = convertService;

		executorService = Utils.newBlockingFixedThreadPool(
				transThreadCnt, 2 * transThreadCnt,
				new NamedThreadFactory("transformService"));
		this.logSender = logSender;
		LOG.debug("MqMessageTransformer init");
	}

	public void destroy(){
		executorService.shutdown();
		try {
			if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
				LOG.error("wait executorService to terminate, timeout");
				executorService.shutdownNow();
			}
		} catch (InterruptedException e) {
			LOG.error("", e);
			executorService.shutdownNow();
		}
	}

	/**
	 * 转换从医院发来的消息, 根据规则做相应的格式转换, 消息转发, 并异步将其发送出去
	 *
	 * @param baseMessage 医院发来的消息
	 * @param messageDispatcher 用于发送消息的 MqMessageDispatcher
	 * @return 用于等待所有转换后的消息发送完成的 CountDownLatch
	 * @throws IOException
     */
	public void transformHospitalMsgAsync(
			T baseMessage, MqMessageDispatcher<T> messageDispatcher)
			throws IOException, CloneNotSupportedException {
		//String resid,String appid,String version

		Map trans = getTransformInfo(baseMessage.getDataType(), baseMessage.getAppServerId(),baseMessage.getDataVersion());
		if(!trans.containsKey("Code") && trans.containsKey("isTransform")){

			baseMessage.setAppSecretKey(trans.get("secretKey").toString());
			baseMessage.setMessageType(BaseMessage.MessageType.RESPONSE);
			//response
			String requestId = baseMessage.getRequestId();
			LOG.debug("requestId   " + requestId);
			String machineID = "";//LWW 用于支持restful 部署多台
			if (requestId.contains("%&"))
			{
				machineID = requestId.split("%&")[0];
				LOG.debug("machineID  " + machineID);
			} else {
				LOG.error("machineID is error ,machineID: " + machineID);
			}
			DispatchRule dispatchRule = new DispatchRule();
			dispatchRule.setAppServerId(baseMessage.getAppServerId());
			dispatchRule.setTargetTopic(Constants.MQ_TO_APPSERVER_TOPIC_PREFIX + machineID);
			dispatchRule.setIsTrans(trans.get("isTransform").toString());
			dispatchRule.setXslContent(trans.get("ruleFileContent").toString());
			dispatchRule.setSecretKey(trans.get("secretKey").toString());
			dispatchRule.setHospitalId(baseMessage.getHospitalId());
			dispatchRule.setHospitalName(baseMessage.getHospitalName());

			LOG.debug("isTransform is " + trans.get("isTransform").toString());
			LOG.debug("ruleFileContent size is " + trans.get("ruleFileContent").toString().length());
			LOG.debug("secretKey is " + trans.get("secretKey").toString());

			executorService.submit(new Task(baseMessage, messageDispatcher, dispatchRule));

		}else{
			//表示查找数据库获取转换所需的信息时存在问题
			LOG.error("getTransformInfo return is error!" + trans.get("Code"));
		}
		}

	@SuppressWarnings("unchecked")
	private T trans(T baseMessage, DispatchRule rule,MqMessageDispatcher<T> messageDispatcher)
			throws IOException, CloneNotSupportedException {

		T retMsg = (T) baseMessage.newCopy();
		try {
			retMsg.setTopic(rule.getTargetTopic());
			retMsg.setAppServerId(rule.getAppServerId());
			retMsg.setAppServerName(rule.getAppServerName());

			if(rule.getIsTrans().equals("true")){//标识需要进行转换

				logSender.messageSentToTransform(retMsg);//日志记录发往规则转换的记录

				RetInfo retInfo = doTransRequest(baseMessage.getFid(),
						baseMessage.getMessageBody(),  baseMessage.getAppSecretKey(),rule.getXslContent(),messageDispatcher);
				retMsg.setMessageBody(retInfo.getContent());

				logSender.messageReceivedFromTransform(retMsg);//表示收到规则转换后的记录

			}else{
				retMsg.setMessageBody(baseMessage.getMessageBody());//直接将院内返回的内容返回
			}
			return retMsg;
		} catch (Exception e) {
			throw new IOException(e);
		}
	}

	private RetInfo doTransRequest(String fid,String oriMsgBody,  String key,String xslContent,MqMessageDispatcher<T> messageDispatcher)
			throws IOException {
		return doTransRequest(fid,oriMsgBody, key, xslContent,2,messageDispatcher);
	}

	private RetInfo doTransRequest(String fid, String oriMsgBody,  String key,String xslContent,
								  int retryCnt,MqMessageDispatcher<T> messageDispatcher) throws IOException {

		RpcException re = null;
		for (int i = 0; i < retryCnt; i++) {
			try {
				if (i != 0) {
					Thread.sleep(3000);
					LOG.debug("doTransRequest-sleep 3秒 ，-i:" + i);
				}
				if(xslContent !=null && !(xslContent.equals(""))){

					HashMap<String,Object> param = new HashMap<String, Object>();
					param.put("fid",fid);
					param.put("oriMsgBody" ,oriMsgBody);
					param.put("xslContent" ,xslContent);
					param.put("key" ,key);
					param.put("convertService",convertService);

                    RetInfo r = messageDispatcher.trans(param);

					return r;
				}else{
					LOG.error("XSLContent is null or empty!xslContent is:" + i);
				}
			} catch (RpcException e){
				LOG.warn("doTransRequest - got RpcException in Dubbo, going to retry: " + i, e);
				re = e;
				continue;
			} catch (Exception e){
				throw new IOException(e);
			}
		}
		LOG.error("still failed to call rpc doTransRequest after " + retryCnt + " times");
		throw new IOException(re);
	}

	public Map<String,String> getTransformInfo(String resid,String appid,String version){

		Map result = dispatchService.getTransformInfo(resid,appid,version);
		return result;
	}

	private class Task implements Runnable {
		private final T message;
		private final MqMessageDispatcher<T> messageDispatcher;
		private final DispatchRule rule;


		public Task(T message, MqMessageDispatcher<T> messageDispatcher,
					DispatchRule rule) {
			this.message = message;
			this.messageDispatcher = messageDispatcher;
			this.rule = rule;

		}

		@Override
		public void run() {
			try{
				T mess= trans(message, rule,messageDispatcher);

				messageDispatcher.sendToAppserver(mess);

				logSender.messageSentToRestful(mess);

			} catch (Exception e) {
				LOG.error("", e);

			}
		}
	}

}
